package com.stxx.stiot2.gateway;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSON;
import com.stxx.framwork.constants.SceneActionType;
import com.stxx.framwork.exception.ServiceException;
import com.stxx.stiot2.domain.Gateway;
import com.stxx.stiot2.domain.ext.ReportData;
import com.stxx.stiot2.event.dto.P2tMsg;
import com.stxx.stiot2.event.dto.T2pMsg;
import com.stxx.stiot2.gateway.face.GatewayServer;
import lombok.extern.slf4j.Slf4j;
import org.dromara.mica.mqtt.codec.MqttProperties;
import org.dromara.mica.mqtt.codec.MqttVersion;
import org.dromara.mica.mqtt.core.client.IMqttClientConnectListener;
import org.dromara.mica.mqtt.core.client.MqttClient;
import org.dromara.mica.mqtt.server.solon.MqttServerTemplate;
import org.noear.solon.Solon;
import org.noear.solon.core.event.EventBus;
import org.tio.core.ChannelContext;

import java.nio.charset.StandardCharsets;
import java.util.UUID;

@Slf4j
public abstract class AbstractMqttClientServer implements GatewayServer {
    MqttClient client = null;
    @Override
    public void init(Gateway gateway){
// 初始化 mqtt 客户端
        client = MqttClient.create()
                .ip(gateway.getParams().getHost())                // mqtt 服务端 ip 地址
                .port(gateway.getParams().getPort())                     // 默认：1883
                .username(gateway.getParams().getUsername())              // 账号
                .password(gateway.getParams().getPassword())             // 密码
                .version(MqttVersion.valueOf(gateway.getParams().getMqttVersion().getValue()))    // 默认：3_1_1
                .clientId(StrUtil.isEmpty(gateway.getParams().getClientId())? UUID.randomUUID().toString().replace("-",""):gateway.getParams().getClientId())             // 非常重要务必手动设置，一般设备 sn 号，默认：MICA-MQTT- 前缀和 36进制的纳秒数
                .readBufferSize(512)            // 消息一起解析的长度，默认：为 8092 （mqtt 消息最大长度）
                .maxBytesInMessage(1024 * 10)   // 最大包体长度,如果包体过大需要设置此参数，默认为： 10M (10*1024*1024)
                .keepAliveSecs(120)             // 默认：60s
                .timeout(10)                    // 超时时间，t-io 配置，可为 null，为 null 时，t-io 默认为 5
                .reconnect(true)                // 是否重连，默认：true
                .reInterval(5000)               // 重连重试时间，reconnect 为 true 时有效，t-io 默认为：5000
                .willMessage(builder -> {
                    builder.topic("/test/offline").messageText("down");    // 遗嘱消息
                })
                .connectListener(new IMqttClientConnectListener() {
                    @Override
                    public void onConnected(ChannelContext context, boolean isReconnect) {
                        log.info("链接服务器成功...");
                    }

                    @Override
                    public void onDisconnect(ChannelContext channelContext, Throwable throwable, String remark, boolean isRemove) {
                        log.info("与链接服务器断开连接...");
                    }
                })
                .properties(MqttProperties.NO_PROPERTIES)                   // mqtt5 properties
                .connectSync();                 // 同步连接，也可以使用 connect()，可以避免 broker 没启动照成启动卡住。

        // 消息订阅，同类方法 subxxx
        client.subQos0(gateway.getParams().getSubTopic(), (context, topic, message, payload) -> {
            T2pMsg t2pMsg = this.receive(topic,new String(payload, StandardCharsets.UTF_8));
            EventBus.publish(t2pMsg);
        });
    }

    @Override
    public T2pMsg receive(String topic, String message){
        String [] strs = topic.split("/");
        String gatewayCode = strs[2];
        String terminalCode = strs[3];
        T2pMsg message2 = new T2pMsg();
        message2.setGatewayCode(gatewayCode);
        message2.setTerminalCode(terminalCode);
        message2.setData(JSON.parseObject(message, ReportData.class));
        return message2;
    }

    @Override
    public void send(String gatewayCode, String terminalCode, P2tMsg message){
        if(client==null){
            throw new ServiceException("网关未初始化或者已关闭");
        }
        if(StrUtil.equals(message.getType(), SceneActionType.READ.getValue())){
            StringBuilder sb = new StringBuilder();
            sb.append("/iot/").append(gatewayCode).append("/").append(terminalCode).append("/properties/get");
            client.publish(sb.toString(),message.getProperties().getBytes(StandardCharsets.UTF_8));
        }else{
            StringBuilder sb = new StringBuilder();
            sb.append("/iot/").append(gatewayCode).append("/").append(terminalCode).append("/properties/set");
            client.publish(sb.toString(),message.getProperties().getBytes(StandardCharsets.UTF_8));
        }
    }

    @Override
    public void close(){
        if(client.isConnected()){
            client.disconnect();
        }
        client.stop();
        client = null;
    }
}
